package org.redis.cluster.redis.support;

import org.redis.cluster.config.ClientConfig;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import redis.clients.jedis.Jedis;

public class RedisSlaveProcessor {
	
	private final static int DEFAULT_SLEEP=1000*1;
	
	private final static int RETRY_COUNT=10;
	
	private final static RedisSlaveProcessor INSTANCE=new RedisSlaveProcessor();
	
	protected final Logger logger=LoggerFactory.getLogger(getClass());
	
	public static RedisSlaveProcessor getInstance() {
		return INSTANCE;
	}
	
	/**
	 * @param slave
	 * @param master
	 * @param isSlaveOfOne
	 * @return
	 */
	public boolean slave(ClientConfig slave,ClientConfig master,boolean isSlaveOfOne) {
		Jedis jedis=getJedis(slave);
		if(jedis==null) {
			return false;
		}
		
		try {
			jedis.slaveof(master.getHost(), master.getPort());
			if(!isSlaveOfOne) {
				return true;
			}
			
			boolean result=waitComplete(jedis);
			jedis.close();
			return result;
		}
		catch(Exception e) {
			if(logger.isDebugEnabled()) {
				logger.debug("",e);
			}
			
			return false;
		}
	}
	
	/**
	 * @param config
	 * @return
	 */
	public boolean slaveofNoOne(ClientConfig config) {
		Jedis jedis=getJedis(config);
		if(jedis==null) {
			return false;
		}
		
		try {
			jedis.slaveofNoOne();
			return true;
		}
		catch(Exception e) {
			if(logger.isDebugEnabled()) {
				logger.debug("",e);
			}
			
			return false;
		}
	}
	
	protected Jedis getJedis(ClientConfig config) {
		try {
			Jedis jedis=new Jedis(config.getHost(),config.getPort(),config.getTimeout());
			jedis.ping();
			
			return jedis;
		}
		catch(Exception e) {
			if(logger.isDebugEnabled()) {
				logger.debug("",e);
			}
			
			return null;
		}
	}
	
	protected boolean waitComplete(Jedis jedis) throws Exception {
		for(int i=0;i<RETRY_COUNT;i++) {
			Thread.currentThread().sleep(DEFAULT_SLEEP);
			
			if(isReplicateComplete(jedis)) {
				jedis.slaveofNoOne();
				return true;
			}
		}
		
		return false;
	}
	
	protected boolean isReplicateComplete(Jedis jedis) {
		String info=jedis.info();
		String lines[]=info.split("\r\n");
		for(String line:lines) {
			String lineSplited[]=line.split(":");
			if(lineSplited.length!=2) {
				continue;
			}
			
			if(lineSplited[0].equals("master_sync_in_progress")) {
				return lineSplited[1].equals("0");
			}
		}
		
		return true;
	}
}